src/openrtc/pool.py (1)

273-275: Prefer a typed event payload for this public API.

Line 273 exposes list[dict[str, Any]], which leaves the stream contract implicit for callers. Consider a TypedDict/dataclass event type and return list[MetricsStreamEvent] here.

♻️ Proposed typing refinement

-from typing import Any, TypeVar
+from typing import Any, TypeVar, TypedDict
+
+class MetricsStreamEvent(TypedDict, total=False):
+    event: str
+    timestamp: float
+    agent: str
+    error: str

-    def drain_metrics_stream_events(self) -> list[dict[str, Any]]:
+    def drain_metrics_stream_events(self) -> list[MetricsStreamEvent]:
         """Drain pending session lifecycle events for JSONL sidecar export."""
         return self._runtime_state.metrics.drain_stream_events()

As per coding guidelines, “Prefer concrete types over Any” and “Prefer well-defined event payload types over loosely shaped dicts.”

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/pool.py` around lines 273 - 275, Change the public API of
drain_metrics_stream_events to return a concrete event type instead of
list[dict[str, Any]]: define a MetricsStreamEvent TypedDict or dataclass
describing the expected keys/values, update the return annotation of
drain_metrics_stream_events to list[MetricsStreamEvent], and propagate that type
to runtime_state.metrics.drain_stream_events() (update its
signature/implementation to return List[MetricsStreamEvent]). Ensure you add the
necessary imports (TypedDict/dataclass and typing.List) and update any
callers/tests to use the new MetricsStreamEvent type.

tests/test_metrics_stream.py (1)

27-50: Consider extracting shared _minimal_snapshot() fixture.

This helper is duplicated in tests/test_tui_app.py. A shared conftest.py fixture would reduce duplication, though this is not blocking.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_metrics_stream.py` around lines 27 - 50, Duplicate helper
_minimal_snapshot() returning PoolRuntimeSnapshot is used across tests; extract
it into a shared pytest fixture in conftest.py. Create a fixture (e.g.,
minimal_snapshot or minimal_snapshot_factory) that constructs and returns the
PoolRuntimeSnapshot (including ProcessResidentSetInfo and SavingsEstimate),
replace local _minimal_snapshot() uses in tests with the injected fixture, and
update imports/usage so tests call the fixture name instead of redefining
_minimal_snapshot().

src/openrtc/resources.py (1)

145-149: Minor: Redundant list() wrapper.

raw_events is already a list from state.get("_stream_events", []), so the list(raw_events) call on line 147 is redundant. It's harmless but could be simplified.

♻️ Optional simplification

         raw_events = state.get("_stream_events", [])
         self._stream_events = deque(
-            list(raw_events),
+            raw_events,
             maxlen=_STREAM_EVENTS_MAXLEN,
         )

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/resources.py` around lines 145 - 149, The code wraps raw_events
in an unnecessary list when constructing self._stream_events; raw_events already
comes from state.get("_stream_events", []), so remove the redundant list(...)
and pass raw_events directly into deque(..., maxlen=_STREAM_EVENTS_MAXLEN)
(update the deque construction around self._stream_events and raw_events
accordingly).

docs/.vitepress/theme/custom.css (1)

5-5: Optional: Stylelint prefers string notation for @import.

The static analysis suggests using a bare string instead of url() for the import. Both are valid CSS, but the string form is more concise.

♻️ Optional style fix

-@import url('https://fonts.googleapis.com/css2?family=DM+Sans:ital,opsz,wght@0,9..40,400..700;1,9..40,500..700&family=JetBrains+Mono:wght@400;500&display=swap');
+@import 'https://fonts.googleapis.com/css2?family=DM+Sans:ital,opsz,wght@0,9..40,400..700;1,9..40,500..700&family=JetBrains+Mono:wght@400;500&display=swap';

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@docs/.vitepress/theme/custom.css` at line 5, Replace the `@import` rule in
docs/.vitepress/theme/custom.css to use string notation instead of url()—locate
the `@import` declaration (the line starting with "@import url(") and change it to
the shorter string form (a quoted URL) so the import uses "@import
'https://fonts.googleapis.com/...';" which satisfies the stylelint suggestion.

src/openrtc/metrics_stream.py (1)

80-83: Document truncation behavior in the docstring.

The open() method truncates the file (mode "w"), which is intentional per the class docstring but could surprise users expecting append behavior. Consider adding a note in the method docstring.

📝 Suggested docstring enhancement

     def open(self) -> None:
-        """Create parent dirs, truncate file, open for append."""
+        """Create parent dirs, truncate file, and open for writing.
+
+        The file is truncated on each worker start so the TUI sees fresh data.
+        """
         self._path.parent.mkdir(parents=True, exist_ok=True)
         self._file = self._path.open("w", encoding="utf-8")

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/metrics_stream.py` around lines 80 - 83, The open() method
currently opens the file with mode "w" (self._path.open("w", encoding="utf-8")),
which truncates the file; update the open() method docstring to explicitly state
that it will truncate existing files (not append) and that this is intentional
per the class behavior so callers are not surprised—mention the use of "w" mode
and reference the open() method and self._path.open call in the docstring.

src/openrtc/tui_app.py (1)

20-27: Consider using TextIOWrapper | None instead of Any for _fh.

The file handle is typed as Any but could use a more precise type like TextIOWrapper | None from typing (or IO[str] | None). This improves IDE support and catches potential misuse.

♻️ Suggested type refinement

+from typing import Any, TextIO
+
 class MetricsTuiApp(App[None]):
     ...
     def __init__(self, watch_path: Path, *, from_start: bool = False) -> None:
         super().__init__()
         self._path = watch_path.resolve()
         self._from_start = from_start
-        self._fh: Any = None
+        self._fh: TextIO | None = None
         self._buf = ""

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/tui_app.py` around lines 20 - 27, Replace the broad Any type on
the instance file handle with a precise text-stream type: update the _fh
annotation in __init__ (and any other attribute or usage sites) from "Any" to
"io.TextIOWrapper | None" or "typing.IO[str] | None"; add the required import
(from io import TextIOWrapper or from typing import IO) and ensure any code that
assigns or reads _fh is compatible (e.g., open(..., "r", encoding=...) returns
TextIOWrapper/IO[str]) so IDE type checks and linting pick up correct usage of
_fh and related operations like reads/writes.

src/openrtc/cli_app.py (2)

389-393: Edge case: value arguments starting with -- would be skipped.

The check not argv_tail[i].startswith("--") assumes values don't start with --. A path like --weird-dir would be incorrectly treated as another flag. This is an unlikely edge case but worth noting.

Consider documenting this limitation or using a more robust parsing approach if paths with -- prefixes are plausible.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/cli_app.py` around lines 389 - 393, The CLI loop that handles
flags in _OPENRTC_ONLY_FLAGS_WITH_VALUE incorrectly treats any next token
beginning with "--" as another flag and skips it; update the logic in the
parsing loop that advances i (the argv_tail index for arg) so that when arg is
in _OPENRTC_ONLY_FLAGS_WITH_VALUE you always consume the next token as the
flag's value regardless of whether it starts with "--" (or alternatively detect
known flags from the flag set rather than using startswith), ensuring
argv_tail[i] (the value) is not accidentally skipped; adjust the handling around
arg, i, and argv_tail to consume the value deterministically and document the
behavior in the parsing function comment.

---

118-173: Consider extracting duplicated scheduling logic.

The dashboard and non-dashboard branches share nearly identical scheduling code (lines 134-152 vs 156-173). This could be extracted into a helper method, though the current implementation is functional and readable.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/cli_app.py` around lines 118 - 173, The _run method contains
duplicated scheduling loops for dashboard and non-dashboard modes; extract that
shared timing/wait/dispatch logic into a helper method (e.g., _scheduling_loop
or _run_scheduling_loop) that accepts an optional live renderable or a flag to
call _build_dashboard_renderable and live.update when provided. Move the shared
computations and checks that use _stop_event, _needs_periodic_file_or_ui,
_jsonl_sink, _refresh_seconds, _jsonl_interval, _write_json_snapshot, and
_emit_jsonl into that helper and replace both duplicated while loops in _run
with a single call to the new helper (ensuring the dashboard branch still wraps
Live and calls the helper with the live object so final
live.update/_build_dashboard_renderable behavior is preserved).

src/openrtc/tui_app.py (1)

90-97: ⚠️ Potential issue | 🟡 Minor

Guard against None wall time to prevent crash on malformed records.

If wall_time_unix is missing or None in a malformed record that passes schema validation, float(wall) will raise TypeError. Add a fallback.

🛡️ Suggested defensive fix

         seq = self._latest.get("seq")
         wall = self._latest.get("wall_time_unix")
         status = self.query_one("#status", Static)
+        wall_val = float(wall) if wall is not None else 0.0
         status.update(
-            f"seq={seq}  wall={float(wall):.3f}  registered={payload.get('registered_agents')} "
+            f"seq={seq}  wall={wall_val:.3f}  registered={payload.get('registered_agents')} "
             f"active={payload.get('active_sessions')}  "
             f"uptime={float(payload.get('uptime_seconds', 0)):.1f}s"
         )

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/openrtc/tui_app.py` around lines 90 - 97, The code calls float(wall)
without guarding against wall being None, which raises TypeError for malformed
records; in the block that reads seq = self._latest.get("seq") and wall =
self._latest.get("wall_time_unix"), ensure wall is given a safe fallback (e.g.,
0.0 or "n/a") before formatting, then use that safe value when building the
status string passed to status.update in query_one("#status", Static); change
the float conversion to operate on the fallback (or conditionally format as
"n/a") so None cannot reach float().

tests/test_metrics_stream.py (1)

188-251: Consider potential flakiness in timing-based tests.

The tests use time.sleep(0.6) and time.sleep(0.85) with a 0.25s interval, expecting 2-3+ emissions. While the assertions use >= 1 and >= 2 to be lenient, on slow CI systems the actual number of emissions might vary.

The current approach is reasonable for this use case since:

1. The intervals are generous (2-3x the expected duration)
2. Assertions check minimum counts rather than exact counts
3. This tests actual time-based behavior that's hard to mock cleanly

If flakiness occurs, consider increasing sleep durations or using synchronization hooks.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_metrics_stream.py` around lines 188 - 251, The timing-based tests
test_runtime_reporter_emits_snapshot_then_drained_events_in_order and
test_runtime_reporter_emits_jsonl_periodically can be flaky on slow CI because
they rely on fixed time.sleep durations versus RuntimeReporter emissions
(metrics_jsonl_interval and refresh_seconds); to fix, make the tests robust by
either increasing the sleep durations (e.g., to cover additional intervals) or,
better, replace the sleeps with a synchronization approach: expose or poll
RuntimeReporter/Pool state (e.g., wait until path contains the expected number
of lines or until seq increments) after reporter.start() and before
reporter.stop(), using a short retry loop with a timeout so tests wait
deterministically for emissions from RuntimeReporter rather than relying on
fragile fixed sleeps.